package org.budo.warehouse.logic.consumer.async;

import java.util.ArrayList;
import java.util.List;

import org.budo.support.servlet.util.QueryStringUtil;
import org.budo.support.spring.util.SpringUtil;
import org.budo.warehouse.logic.api.DataConsumer;
import org.budo.warehouse.logic.api.DataEntry;
import org.budo.warehouse.logic.api.DataMessage;
import org.budo.warehouse.logic.consumer.DataEntryPojo;
import org.budo.warehouse.logic.producer.DataMessageImpl;
import org.budo.warehouse.logic.util.PipelineUtil;
import org.budo.warehouse.service.api.IDataNodeService;
import org.budo.warehouse.service.entity.DataNode;
import org.budo.warehouse.service.entity.Pipeline;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

/**
 * 非IocBean
 * 
 * @author limingwei
 */
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class AsyncDataEntryConsumerWrapper implements DataConsumer {
    private DataConsumer dataConsumer;

    private Pipeline pipeline;

    @Override
    public void consume(DataMessage dataMessage) {
        DataMessageImpl dataMessagePojo = this.toPojo(dataMessage);

        IDataNodeService dataNodeService = SpringUtil.getBean(IDataNodeService.class);
        DataNode targetDataNode = dataNodeService.findByIdCached(this.getPipeline().getTargetDataNodeId());

        // 旧版消息格式
        String v1 = QueryStringUtil.getParameter(targetDataNode.getUrl(), "v1", "false");
        if ("true".equals(v1)) {
            V1AsyncMessageSender sender = SpringUtil.getBean(V1AsyncMessageSender.class);
            sender.send_message_v1(this.getPipeline(), dataMessagePojo);
            return;
        }

        // 新的通用的消息格式
        dataConsumer.consume(dataMessagePojo);
    }

    private DataMessageImpl toPojo(DataMessage dataMessage) {
        List<DataEntry> dataEntries = dataMessage.getDataEntries();
        List<DataEntry> dataEntryPojos = new ArrayList<DataEntry>();

        for (DataEntry dataEntry : dataEntries) {
            String targetSchema = PipelineUtil.targetSchema(this.getPipeline(), dataEntry);
            String targetTable = PipelineUtil.targetTable(this.getPipeline(), dataEntry);

            DataEntryPojo dataEntryPojo = new DataEntryPojo(dataEntry) // 序列化传输
                    .setSchemaName(targetSchema) // 供下一步匹配
                    .setTableName(targetTable);

            dataEntryPojos.add(dataEntryPojo);
        }

        return new DataMessageImpl(pipeline.getTargetDataNodeId(), dataEntryPojos);
    }
}